GET /api/v1/conflicts 會列出每一個人的任務)src/api_main.py 的 jobs 字典裡,伺服器一重啟就清空今天先打好地基:資料庫模型、密碼雜湊、JWT 登入。明天 Day 16 再把檢測端點接上登入與資料庫。
第 2 週:檢測優化與生產系統
今天要完成:
src/database.py:SQLAlchemy 引擎、Session、User 與 Job 兩個模型src/auth.py:PBKDF2-SHA256 + 隨機 salt,不儲存明文密碼src/auth.py:登入後簽發 access token,後續請求帶著它證明身份src/api_auth.py:/api/v1/auth/register、/login、/me、/logout
tests/test_day15_auth.py:6 個測試,包含一個今天修掉的郵箱大小寫漏洞資料庫總有外洩的可能(備份檔、SQL injection、內部人員)。如果存的是明文,外洩當下所有帳號就都失守了;而且很多人在不同網站用同一組密碼。
所以只存密碼的雜湊值:登入時把使用者輸入的密碼用同樣方式雜湊,再比對結果。但一般的 SHA-256 太快了,攻擊者每秒可以試上億組密碼。密碼雜湊需要兩個額外設計:
登入成功後,伺服器要記得「這個請求是誰發的」。傳統做法是 session:伺服器存一張「session id → 用戶」的表。JWT(JSON Web Token)則是把用戶 id 和到期時間寫進 token,再用只有伺服器知道的 SECRET_KEY 簽章:
header.payload.signature
payload = {"sub": "<user_id>", "exp": <到期時間>}
伺服器收到 token 時只要驗證簽章與到期時間,不必查表。代價是 token 一旦簽發,在到期前都無法作廢,後面驗證結果會看到這一點。
POST /api/v1/auth/register {email, password, full_name}
→ normalize_email → hash_password → 寫入 users 表
POST /api/v1/auth/login username=<email>&password=...(表單格式)
→ authenticate_user(查 email、verify_password)→ create_access_token → 回傳 JWT
GET /api/v1/auth/me Authorization: Bearer <JWT>
→ get_current_user(verify_token → 查 user)→ 回傳用戶資料
users 表中的帳號、access token、目前登入的用戶src/database.py、src/auth.py、src/api_auth.py(新增);src/api_main.py(修改:掛上路由、建立資料表)/api/v1/detect 會用 get_current_user 保護端點,並把任務寫進 jobs 表專案結構變化:
srs-review-agent/
├── src/
│ ├── api_main.py ← 修改:include_router(auth_router)、init_db()
│ ├── database.py ← 【新增】引擎、Session、User、Job
│ ├── auth.py ← 【新增】密碼雜湊、JWT、用戶查詢
│ ├── api_auth.py ← 【新增】認證路由與 get_current_user
│ └── ...
├── tests/
│ └── test_day15_auth.py ← 【新增】6 個測試
└── test.db ← 開發用 SQLite(執行後自動產生)
pip install sqlalchemy "python-jose[cryptography]" email-validator python-multipart
| 套件 | 用途 |
|---|---|
sqlalchemy |
ORM,把 Python 類別對應到資料表 |
python-jose[cryptography] |
簽發與驗證 JWT |
email-validator |
Pydantic 的 EmailStr 需要它來檢查郵箱格式 |
python-multipart |
登入端點用表單格式接收帳密,FastAPI 需要它解析表單 |
這些都已列在 requirements.txt。另外有兩個環境變數:
| 變數 | 預設 | 說明 |
|---|---|---|
DATABASE_URL |
sqlite:///./test.db |
改成 postgresql://... 即可換資料庫 |
SECRET_KEY |
dev-secret-key-change-in-production |
JWT 簽章金鑰,正式環境一定要換 |
建立 src/database.py:
from sqlalchemy import create_engine, Column, String, Boolean, DateTime, JSON, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
import uuid
import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
if "sqlite" in DATABASE_URL:
# SQLite 預設禁止跨執行緒使用同一個連線;FastAPI 會在不同執行緒處理請求
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
else:
engine = create_engine(DATABASE_URL, echo=False)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
"""FastAPI 依賴:每個請求一個 Session,結束時關閉"""
db = SessionLocal()
try:
yield db
finally:
db.close()
get_db() 是一個產生器,搭配 FastAPI 的 Depends 使用:請求進來時建立 Session,回應送出後執行 finally 關閉連線。
./test.db 是相對於啟動目錄的路徑。在專案根目錄啟動 uvicorn,資料庫就會建在專案根目錄。
同一個檔案中定義兩張表:
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
full_name = Column(String)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
jobs = relationship("Job", back_populates="user", cascade="all, delete-orphan")
class Job(Base):
__tablename__ = "jobs"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
user_id = Column(String, ForeignKey("users.id"))
status = Column(String, default="processing") # processing, completed, failed
constraints = Column(JSON) # 原始約束
results = Column(JSON, nullable=True) # 檢測結果(與 Day 13 API 的 results 同格式)
error = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
completed_at = Column(DateTime, nullable=True)
user = relationship("User", back_populates="jobs")
def init_db():
"""建立所有資料表(已存在則略過)"""
Base.metadata.create_all(bind=engine)
幾個設計決定:
email 設 unique=True:資料庫層級保證不重複,是應用層檢查之外的第二道防線constraints、results 用 JSON 欄位:檢測結果的結構就是 Day 13 API 回傳的 {"conflicts_found": ..., "conflicts": [...]},直接整包存,不用再拆成多張表Job 今天只定義、還不使用:Day 16 的檢測端點才會寫入它init_db() 用 create_all 建表,只會建立不存在的表,不會修改已存在的表。之後如果要加欄位,需要 Alembic 這類遷移工具,這裡先不處理。
src/api_main.py 在啟動時呼叫它,並掛上今天的路由:
from src.database import init_db
from src.api_auth import router as auth_router
try:
init_db()
except Exception as e:
print(f"⚠️ 數據庫初始化: {e}")
app.include_router(auth_router)
建立 src/auth.py:
import hashlib
import hmac
import os
def hash_password(password: str) -> str:
"""加密密碼使用 PBKDF2-SHA256"""
salt = os.urandom(16).hex()
pwdhash = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'),
bytes.fromhex(salt), 100000)
return f"pbkdf2_sha256${salt}${pwdhash.hex()}"
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""驗證密碼"""
try:
if not hashed_password.startswith("pbkdf2_sha256$"):
return False
_, salt, pwdhash = hashed_password.split("$", 2)
test_hash = hashlib.pbkdf2_hmac('sha256', plain_password.encode('utf-8'),
bytes.fromhex(salt), 100000)
return hmac.compare_digest(test_hash.hex(), pwdhash)
except Exception:
return False
hashlib 就有,不需要額外套件。bcrypt、Argon2 也是常見選擇,需要另外安裝演算法$salt$雜湊值:salt 必須和雜湊值一起存,驗證時才能用同一組 salt 重算;開頭的演算法名稱讓日後換演算法時能分辨新舊格式hmac.compare_digest:固定時間比對。一般的 == 遇到第一個不同的字元就回傳,攻擊者可以從回應時間推測猜對了幾個字元實際存進資料庫的樣子:
pbkdf2_sha256$5ae2e2101c38fe8f0abee415fe533e23$35bcc5b06818e9c5a5e95cdea36fad17...
同一個檔案中:
from datetime import datetime, timedelta
from jose import JWTError, jwt
from typing import Optional
SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> Optional[dict]:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
return None
return {"user_id": user_id}
except JWTError:
return None
jwt.decode 會同時檢查簽章與 exp。簽章不符、格式錯誤、已過期都會丟出 JWTError,統一回傳 None,由呼叫端決定回 401。
access token 只有 15 分鐘效期。時間短,token 外洩時能被濫用的時間就短;但使用者也得頻繁重新登入。Day 17 會用 refresh token 解決這個矛盾。
同一個檔案還有查詢用戶的函數:
def normalize_email(email: str) -> str:
"""統一郵箱格式:去除空白並轉小寫"""
return email.strip().lower()
def authenticate_user(db, email: str, password: str):
from src.database import User
user = db.query(User).filter(User.email == normalize_email(email)).first()
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
normalize_email() 是測試時補上的。原本的程式直接用使用者輸入的郵箱查詢與儲存,實測發現:
註冊 alice@example.com → 200
註冊 ALICE@Example.com → 200(又建了一個帳號!)
用 Alice@example.com 登入 alice 的帳號 → 401
Pydantic 的 EmailStr 只會把網域轉小寫(ALICE@Example.com → ALICE@example.com),@ 前面維持原樣。結果同一個郵箱能註冊兩個帳號,而使用者只是大小寫打錯就登不進去。現在註冊(src/api_auth.py)、authenticate_user、get_user_by_email 都先經過 normalize_email()。
建立 src/api_auth.py,先定義路由與「取得目前用戶」的依賴:
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
router = APIRouter(prefix="/api/v1/auth", tags=["authentication"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/auth/login")
async def get_current_user(token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)) -> User:
payload = verify_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="無效的認證令牌",
headers={"WWW-Authenticate": "Bearer"})
user = get_user_by_id(db, payload.get("user_id"))
if user is None:
raise HTTPException(status_code=401, detail="用戶不存在")
if not user.is_active:
raise HTTPException(status_code=403, detail="用戶已被停用")
return user
OAuth2PasswordBearer 會從 Authorization: Bearer <token> 標頭取出 token;沒有標頭時直接回 401 Not authenticated。任何端點只要加上 current_user: User = Depends(get_current_user),就變成需要登入。
tokenUrl 告訴 /docs 的 Swagger UI 去哪裡登入,所以可以直接在瀏覽器按「Authorize」測試受保護的端點。
註冊端點:
class UserCreate(BaseModel):
email: EmailStr
password: str
full_name: str
@router.post("/register", response_model=UserResponse)
async def register(user_data: UserCreate, db: Session = Depends(get_db)):
if len(user_data.password) < 8:
raise HTTPException(status_code=400, detail="密碼至少需要 8 個字符")
if get_user_by_email(db, user_data.email):
raise HTTPException(status_code=400, detail="郵箱已被註冊")
new_user = User(id=str(uuid.uuid4()),
email=normalize_email(user_data.email),
hashed_password=hash_password(user_data.password),
full_name=user_data.full_name)
db.add(new_user)
db.commit()
db.refresh(new_user)
return UserResponse.model_validate(new_user)
response_model=UserResponse 限定回應只包含 id、email、full_name、is_active,即使 User 物件上有 hashed_password,也不會出現在回應中。UserResponse 設定了 from_attributes = True,model_validate 才能從 SQLAlchemy 物件讀取屬性。
登入端點:
@router.post("/login", response_model=LoginResponse)
async def login(form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)):
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="郵箱或密碼錯誤",
headers={"WWW-Authenticate": "Bearer"})
access_token = create_access_token(
data={"sub": user.id},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
return {"access_token": access_token, "token_type": "bearer",
"user": UserResponse.model_validate(user)}
兩個要注意的地方:
username(填郵箱)與 password。這是 OAuth2 password flow 的規定,送 JSON 會得到 422/me 與 /logout 都只依賴 get_current_user,完整代碼見 src/api_auth.py。src/api_auth.py 裡的 /jobs、/jobs/{job_id} 兩個端點屬於 Day 16 的任務隔離,明天再說明。
cd srs-review-agent
python3 -m pytest tests/test_day15_auth.py -v
tests/test_day15_auth.py::test_password_hashing PASSED [ 16%]
tests/test_day15_auth.py::test_jwt_token PASSED [ 33%]
tests/test_day15_auth.py::test_token_expiration PASSED [ 50%]
tests/test_day15_auth.py::test_database_models PASSED [ 66%]
tests/test_day15_auth.py::test_auth_flow PASSED [ 83%]
tests/test_day15_auth.py::test_email_normalization PASSED [100%]
============================== 6 passed in 0.29s ===============================
前 5 個是單元測試(雜湊、JWT、過期、建表、完整流程);test_email_normalization 透過 TestClient 實際呼叫註冊與登入端點,確認大小寫不同的郵箱無法重複註冊、也能正常登入。
注意:
test_database_models與test_email_normalization會寫入預設的./test.db。不想動到開發資料庫時,可以先設定DATABASE_URL=sqlite:////tmp/test_day15.db。
啟動伺服器(uvicorn src.api_main:app --port 8000)後:
1. 註冊
curl -X POST http://localhost:8000/api/v1/auth/register \
-H "Content-Type: application/json" \
-d '{"email": "alice@example.com", "password": "SecurePass123", "full_name": "Alice"}'
{"id": "eb9c4e38-cb28-4072-a674-88d7418585c5", "email": "alice@example.com", "full_name": "Alice", "is_active": true}
回應中沒有密碼雜湊值。
2. 登入(表單格式)
curl -X POST http://localhost:8000/api/v1/auth/login \
-d "username=alice@example.com&password=SecurePass123"
{"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJlYjljNGUzOC1jYjI4LTQwNzItYTY3NC04OGQ3NDE4NTg1YzUiLCJleHAiOjE3OTA2NjI3ODR9.PzmJYQ0L2gQWNx2TsSiE4els_8I68-7tDhUI784YewY",
"token_type": "bearer",
"user": {"id": "eb9c4e38-cb28-4072-a674-88d7418585c5", "email": "alice@example.com", "full_name": "Alice", "is_active": true}}
把 token 中間那段做 Base64 解碼,就能看到內容:
{"sub": "eb9c4e38-cb28-4072-a674-88d7418585c5", "exp": 1790662784}
可見 JWT 的 payload 只是編碼、不是加密,任何人都讀得到,所以不能放密碼或其他敏感資料。簽章保證的是「內容沒被竄改」,不是「內容保密」。
3. 用 token 取得目前用戶
curl http://localhost:8000/api/v1/auth/me -H "Authorization: Bearer <access_token>"
{"id": "eb9c4e38-cb28-4072-a674-88d7418585c5", "email": "alice@example.com", "full_name": "Alice", "is_active": true}
4. 錯誤情況
| 請求 | 狀態碼 | detail |
|---|---|---|
重複註冊 alice@example.com |
400 | 郵箱已被註冊 |
密碼 short |
400 | 密碼至少需要 8 個字符 |
郵箱 not-an-email |
422 | value is not a valid email address: An email address must have an @-sign. |
| 密碼錯誤 | 401 | 郵箱或密碼錯誤 |
| 用 JSON 送登入資料 | 422 | username、password Field required |
/me 不帶 token |
401 | Not authenticated |
/me 帶偽造 token |
401 | 無效的認證令牌 |
5. 登出之後呢?
curl -X POST http://localhost:8000/api/v1/auth/logout -H "Authorization: Bearer <access_token>"
# {"message": "已登出"}
curl http://localhost:8000/api/v1/auth/me -H "Authorization: Bearer <access_token>"
# 200,仍然回傳 Alice 的資料
登出後同一個 token 還是能用。伺服器不記錄已簽發的 token,/logout 其實什麼都沒做,只是提示前端刪掉 token。token 外洩時,在 15 分鐘到期之前都無法撤銷。這是 JWT 無狀態設計的代價,Day 17 討論會話管理時會再回到這個問題。
DATABASE_URL 可換成 PostgreSQL,需另外安裝 psycopg2-binary;本文只在 SQLite 上實測create_all 不會修改既有的表,欄位變更需要 AlembicSECRET_KEY 有預設值:方便開發,但正式環境忘記設定時,任何人都能用這個公開的預設值偽造 token。更安全的做法是正式環境缺少設定時直接拒絕啟動datetime.utcnow() 在 Python 3.12 以後已棄用:目前仍可運作但會出現警告,之後可改用 datetime.now(timezone.utc)
git add src/database.py src/auth.py src/api_auth.py src/api_main.py tests/test_day15_auth.py
git commit -m "Day 15: SQLAlchemy 模型、PBKDF2 密碼雜湊、JWT 註冊與登入"
test.db 是開發用的資料庫檔案,不要提交。
帳號系統有了,但 Day 13 的檢測端點還是誰都能用,任務也還存在記憶體裡。明天 Day 16 會新增需要登入的 /api/v1/detect,把任務寫進今天建立的 jobs 表,讓每個用戶只看得到自己的任務,並在前端加上登入畫面。